package impl

import (
	"context"
	"errors"
	"sync"
	"time"

	"github.com/golang/protobuf/proto"
	"go.uber.org/atomic"
	"hundsun.com/hsl/hschain/blockchain/common"
	"hundsun.com/hsl/hschain/blockchain/impl/pubsub"
	"hundsun.com/hsl/hschain/common/cache"
	"hundsun.com/hsl/hschain/common/collections/fifo"
	"hundsun.com/hsl/hschain/common/collections/listmap"
	"hundsun.com/hsl/hschain/common/config"
	"hundsun.com/hsl/hschain/common/log/logging"
	"hundsun.com/hsl/hschain/common/types"
	"hundsun.com/hsl/hschain/common/util/pb"
	"hundsun.com/hsl/hschain/module/common/utils"
	commonproto "hundsun.com/hsl/hschain/protos/common"
	protosconfig "hundsun.com/hsl/hschain/protos/config"
	"hundsun.com/hsl/hschain/protos/ledger"
	"hundsun.com/hsl/hschain/protos/p2p"
	"hundsun.com/hsl/hschain/store"
	"hundsun.com/hsl/hschain/store/mq"
	"hundsun.com/hsl/hschain/store/state"
)

var (
	// fetchPeerListInterval 同步节点列表频率
	fetchPeerListInterval = &utils.IntModuleConfigParam{Name: "fetchPeerListInterval", DefaultValue: 1}

	// blockSyncInterval 同步区块的频率
	blockSyncInterval = &utils.IntModuleConfigParam{Name: "blockSyncInterval", DefaultValue: 1}

	// ntpSyncInterval ntp时钟同步频率
	ntpSyncInterval = &utils.IntModuleConfigParam{Name: "ntpSyncInterval", DefaultValue: 60}

	// blockCacheNumLimit 区块缓存数量限制
	blockCacheNumLimit = &utils.IntModuleConfigParam{Name: "blockCacheNumLimit", DefaultValue: 1000}

	// maxFetchBlockNum 最大同步区块数
	maxFetchBlockNum = &utils.IntModuleConfigParam{Name: "maxFetchBlockNum", DefaultValue: 1000}

	// enableSaveExecutedTx
	enableSaveExecutedTx = &utils.BoolModuleConfigParam{Name: "enableSaveExecutedTx", DefaultValue: true}

	// workerExecTimeout
	workerExecTimeout = &utils.IntModuleConfigParam{Name: "workerExecTimeout", DefaultValue: DefaultWorkerExecTimeout}
)

// GoroutineErrorWrapper goroutine错误包装
type GoroutineErrorWrapper struct {
	Error        error
	GoroutineTag string
}

// BlockChain blockchain
type BlockChain struct {
	log       logging.Log
	ctx       context.Context
	CancelFun context.CancelFunc

	blockStore *BlockStore

	topicClient  mq.Client
	workerClient mq.Client
	blockCache   *cache.Cache

	// 同步相关配置
	fetchPeerListInterval time.Duration
	blockSyncInterval     time.Duration
	ntpSyncInterval       time.Duration

	// 区块是否已同步：最近验证的块的序列号需要与序列号中心保持一致；
	blockSynchronized *atomic.Bool
	isClosed          *atomic.Bool
	isNtpClockSynched *atomic.Bool

	peerList *p2p.PeerInfos

	//cfg
	maxFetchBlockNum   uint64 //一次最多申请获取block个数
	TimeoutSeconds     uint64
	blockCacheNumLimit uint64
	enablePreCache     bool

	curParam *protosconfig.SysParam

	//boss cfg
	EnableSaveExecutedTx bool // 使能保存已执行交易
	WorkerExecTimeout    uint32

	//boss
	errChan           chan<- GoroutineErrorWrapper
	curBlockNum       int64                     // 当前处理区块
	curRecvBlockNum   int64                     // 当前已经接收到的最大区块编号
	historyTxs        *TxHistoryMap             // 历史区块交易 map[uint64]map[string]bool
	blocksCh          chan *ledger.Block        // Blocks通道
	txGroupsCh        chan *types.BlockTxGroups // 分组通道
	execRstCache      *listmap.ListMap          // 本地缓冲，缓存ExecutedBlock
	blocksAcceptMutex sync.Mutex                // 区块同步锁，如果正在同步区块，mq收到新的区块通知消息时需要等待上一次的同步完成
	txRecordNum       uint32

	stateHashCache *fifo.FIFO  // 已经提交的状态hash缓存
	stateDBHandle  state.State // 状态数据处理
	//msgWg          *sync.WaitGroup
	heartbeatMgr *HeartbeatMgr          // worker心跳管理
	sysparam     *pubsub.SystemParamMgr // 系统参数管理
	pubsubMgr    *pubsub.PubSub         // 订阅者管理
}

// NewBlockChain 新建blockchain
func NewBlockChain(ctx context.Context, cfg *config.ModuleConfig, cnlfn context.CancelFunc, log logging.Log) *BlockChain {
	tClient, err := store.NewMqClient(mq.ModeTopic.String())
	if err != nil {
		panic(err)
	}
	err = tClient.Sub(types.BlockChainModule, mq.ModeTopic.String())
	if err != nil {
		panic(err)
	}

	wClient, err := store.NewMqClient(mq.ModeWorker.String())
	if err != nil {
		panic(err)
	}
	err = wClient.Sub(types.BlockChainModule, mq.ModeWorker.String())
	if err != nil {
		panic(err)
	}

	//boss
	//txRecordNum初始化
	payLoad := pb.GetGenesisTransactionPayload()
	txRecordNum := (payLoad.GetTimeLimit().MaxTxTTL+payLoad.GetTimeLimit().MaxTimeIntervalError*2)*2 + 30
	if txRecordNum < uint32(MaxTxDupCheckBlock) {
		txRecordNum = uint32(MaxTxDupCheckBlock)
	}

	blockstore := NewBlockStore(ctx, log)
	sdl, err := store.NewState(blockstore.stateDB)
	if err != nil {
		panic(err)
	}

	chain := &BlockChain{
		log:                log,
		ctx:                ctx,
		CancelFun:          cnlfn,
		topicClient:        tClient,
		workerClient:       wClient,
		blockStore:         blockstore,
		blockCache:         cache.New(),
		blockCacheNumLimit: uint64(blockCacheNumLimit.GetValue(cfg, log)),
		maxFetchBlockNum:   uint64(maxFetchBlockNum.GetValue(cfg, log)),
		//boss
		EnableSaveExecutedTx: enableSaveExecutedTx.GetValue(cfg, log),
		WorkerExecTimeout:    uint32(workerExecTimeout.GetValue(cfg, log)),

		blockSynchronized:     atomic.NewBool(false),
		isClosed:              atomic.NewBool(false),
		isNtpClockSynched:     atomic.NewBool(false),
		peerList:              &p2p.PeerInfos{},
		fetchPeerListInterval: time.Second * time.Duration(fetchPeerListInterval.GetValue(cfg, log)),
		blockSyncInterval:     time.Second * time.Duration(blockSyncInterval.GetValue(cfg, log)),
		ntpSyncInterval:       time.Second * time.Duration(ntpSyncInterval.GetValue(cfg, log)),
		enablePreCache:        false,

		//boss初始化
		errChan:         make(chan GoroutineErrorWrapper, 1),
		curBlockNum:     -1,
		curRecvBlockNum: -1,
		historyTxs: &TxHistoryMap{
			hash2BlockNum:  make(map[string]uint64),
			blockNum2Hashs: make(map[uint64][]string),
		},
		blocksCh:          make(chan *ledger.Block, MaxAccpetBlock),
		txGroupsCh:        make(chan *types.BlockTxGroups, MaxAccpetBlock),
		execRstCache:      listmap.New(),
		blocksAcceptMutex: sync.Mutex{},
		txRecordNum:       txRecordNum,
		stateHashCache:    fifo.NewFIFO(DefaultStateHashCache),
		stateDBHandle:     sdl,
		heartbeatMgr:      NewHeartbeatMgr(),
		pubsubMgr:         pubsub.NewPubSub(log, blockstore.ledgerDB, tClient, wClient),
		sysparam:          pubsub.NewSystemParamMgr(blockstore.ledgerDB),
	}
	return chain
}

// SaveCurParam save current param
func (chain *BlockChain) SaveCurParam() {
	err := chain.blockStore.SaveCurParam(chain.GetCurParam())
	if err != nil {
		panic(err)
	}
}

// LoadCurParam load current param
func (chain *BlockChain) LoadCurParam() error {
	if chain.blockStore != nil {
		sysParam, err := chain.blockStore.LoadCurParam()
		if err != nil {
			return err
		}
		chain.fillCurParam(sysParam)
	}
	return nil
}

func (chain *BlockChain) fillCurParam(curParam *protosconfig.SysParam) {
	if curParam != nil {
		if curParam.TimeLimit != nil {
			chain.curParam.TimeLimit = curParam.TimeLimit
		}
		if curParam.Difficulty != 0 {
			chain.curParam.Difficulty = curParam.Difficulty
		}
		if curParam.Reward != 0 {
			chain.curParam.Reward = curParam.Reward
		}
		if curParam.GasLimit != 0 {
			chain.curParam.GasLimit = curParam.GasLimit
		}
		if curParam.Crypto != nil {
			chain.curParam.Crypto = curParam.Crypto
		}
		if curParam.CheckBlock != nil {
			chain.curParam.CheckBlock = curParam.CheckBlock
		}
		if curParam.BlockLimit != nil {
			chain.curParam.BlockLimit = curParam.BlockLimit
		}
		if curParam.FeeLimit != nil {
			chain.curParam.FeeLimit = curParam.FeeLimit
		}
		if curParam.Manager != nil {
			chain.curParam.Manager = curParam.Manager
		}
		if curParam.Consensus != nil {
			chain.curParam.Consensus = curParam.Consensus
		}
	}

}

// InitCache 初始化缓存
func (chain *BlockChain) InitCache(height uint64) {
	initNum := height - chain.blockCacheNumLimit
	if initNum < 0 {
		initNum = 0
	}
	for ; initNum <= height; initNum++ {
		executedBlock, err := chain.GetExecutedBlock(initNum)
		if err != nil {
			panic(err)
		}
		chain.blockCache.Set(common.GenTempExecutedBlockKeyByHeight(executedBlock.Block.Header.Number), executedBlock)
	}
}

// InitGenesisBlock 初始化创世块
func (chain *BlockChain) InitGenesisBlock() error {
	genesisBlock := pb.GetGenesisBlock()
	if genesisBlock != nil {
		var blkEntry = &common.BlockEntry{}
		blkEntry.Block = genesisBlock
		chain.blockCache.Set(common.GenTempBlockEntryKeyByHeight(0), blkEntry)
		err := chain.blockStore.SaveGenesisBlock(genesisBlock)
		if err != nil {
			return err
		}
		return nil
	}
	return errors.New("genesis block is nil")
}

// InitSysParam ...
func (chain *BlockChain) InitSysParam() error {
	genesisTransactionPayload := pb.GetGenesisTransactionPayload()
	if genesisTransactionPayload != nil {
		var sysParam = &protosconfig.SysParam{}
		sysParam.Difficulty = genesisTransactionPayload.Difficulty
		sysParam.TimeLimit = genesisTransactionPayload.TimeLimit
		sysParam.Crypto = genesisTransactionPayload.Crypto
		sysParam.CheckBlock = genesisTransactionPayload.CheckBlock
		sysParam.Manager = genesisTransactionPayload.Manager
		sysParam.BlockLimit = genesisTransactionPayload.BlockLimit
		sysParam.GasLimit = genesisTransactionPayload.GasLimit
		sysParam.Reward = genesisTransactionPayload.Reward
		sysParam.FeeLimit = genesisTransactionPayload.FeeLimit

		consensusType := genesisTransactionPayload.Consensus.GetType()
		if consensusType == protosconfig.SoloConsensusType {
			var soloParam = &protosconfig.ConsensusParam_SoloPayload{SoloPayload: &protosconfig.SoloConsensusConfigPayload{}}
			err := proto.Unmarshal(genesisTransactionPayload.Consensus.GetPayload(), soloParam.SoloPayload)
			if err != nil {
				panic(err.Error())
			}
			sysParam.Consensus = &protosconfig.ConsensusParam{}
			sysParam.Consensus.Type = protosconfig.SoloConsensusType
			sysParam.Consensus.ConsensusPayload = soloParam
		} else if consensusType == protosconfig.DposConsensusType {
			var dposParam = &protosconfig.ConsensusParam_DposPayload{DposPayload: &protosconfig.DposConsensusConfigPayload{}}
			err := proto.Unmarshal(genesisTransactionPayload.Consensus.GetPayload(), dposParam.DposPayload)
			if err != nil {
				panic(err.Error())
			}
			sysParam.Consensus = &protosconfig.ConsensusParam{}
			sysParam.Consensus.Type = protosconfig.DposConsensusType
			sysParam.Consensus.ConsensusPayload = dposParam
		} else {
			return errors.New("unknown consensus type")
		}
		chain.curParam = sysParam
	}

	err := chain.LoadCurParam()
	if err != nil {
		panic(err)
	}

	chain.log.Info("init blockchain sysparam successfully")
	return nil
}

// GetCurParam ...
func (chain *BlockChain) GetCurParam() *protosconfig.SysParam {
	return chain.curParam
}

func (chain *BlockChain) subMsg(router *commonproto.Router) error {
	//封装订阅消息
	data := &commonproto.ReqSubscriber{
		MsgRouter: router,
		Type:      pubsub.SubSysParam,
	}
	msg := chain.workerClient.NewMessage(types.BlockChainModule, mq.ModeWorker.String(), types.EventSubscriber, data)
	err := chain.workerClient.Send(msg, false)
	if err != nil {
		return err
	}
	return nil
}

// SynRoutine 主要循环
func (chain *BlockChain) SynRoutine() {
	//获取peerlist的定时器，默认1s
	fetchPeerListTicker := time.NewTicker(chain.fetchPeerListInterval)

	//向peer请求同步block的定时器，默认1s
	blockSynTicker := time.NewTicker(chain.blockSyncInterval)

	// ntp时钟同步频率， 默认1分钟
	ntpSyncTicker := time.NewTicker(chain.ntpSyncInterval)

	for {
		select {
		case <-fetchPeerListTicker.C:
			go chain.FetchPeerList()
		case <-blockSynTicker.C:
			go chain.SyncBlocksFromPeers()
		case <-ntpSyncTicker.C:
			go chain.SyncNtp()

		case <-chain.ctx.Done():
			chain.log.Debugf("SynRoutine task exit.")
			return
		}
	}
}

// ProcCheckedCacheBlock 处理校验后缓存区块
func (chain *BlockChain) ProcCheckedCacheBlock() {
	for {
		select {
		case <-time.After(time.Millisecond * 10):
			nextBlockHeight := chain.GetBlockHeight() + 1
			tempBlkKey := common.GenTempBlockEntryKeyByHeight(nextBlockHeight)
			if entry, ok := chain.blockCache.Get(tempBlkKey); ok {
				blkEntry := entry.(*common.BlockEntry)
				if blkEntry.Checked {
					blkNum := blkEntry.Block.Header.Number
					batch := chain.blockStore.ledgerDB.NewBatch(true)

					err := chain.blockStore.SaveBlock(blkEntry.Block, batch)
					if err != nil {
						chain.log.Errorf("Save block %d failed, err: %s", blkNum, err.Error())
						break
					}

					err = chain.blockStore.SaveBlockNumber(blkNum, batch)
					if err != nil {
						chain.log.Errorf("Save block number %d failed, err: %s", blkNum, err.Error())
						break
					}

					blkSequence := blkEntry.Block.Header.BlockSequence
					err = chain.blockStore.SaveBlockSequence(blkSequence, batch)
					if err != nil {
						chain.log.Errorf("save block sequence %d failed, err: %s", blkSequence, err.Error())
						break
					}

					err = batch.Write()
					if err != nil {
						chain.log.Errorf("Batch save block info failed, err: %s", err.Error())
						break
					}

					chain.log.Infof("Save checked block successfully, blockstore height update from %d to %d", chain.blockStore.height.Load(), blkNum)
					chain.blockStore.height.Store(blkNum)
					chain.blockCache.Delete(tempBlkKey)
					go chain.OnBlockAccepted(blkNum)
				}
			}
		case <-chain.ctx.Done():
			return
		}
	}
}

// ProcWorkerMsg 处理worker模式消息
func (chain *BlockChain) ProcWorkerMsg() {
	for msg := range chain.workerClient.Recv() {
		chain.log.Debugf("blockchain worker client recv msg: %s[%d]", types.GetEventName(int(msg.Ty)), msg.Ty)
		msgtype := msg.Ty
		switch msgtype {
		case types.EventTxQuery:
			go mq.ProcessMsg(msg, chain.queryTx, nil, chain.log)
		case types.EventIsSynched:
			go mq.ProcessMsg(msg, chain.isSynched, nil, chain.log)
		case types.EventGetBlock:
			go mq.ProcessMsg(msg, chain.getBlock, nil, chain.log)
		case types.EventRequestBlock:
			go mq.ProcessMsg(msg, chain.requestBlock, nil, chain.log)
		case types.EventGetBlocks:
			go mq.ProcessMsg(msg, chain.getBlocks, nil, chain.log)
		case types.EventGetBlockExecResult:
			go mq.ProcessMsg(msg, chain.getBlockExecResult, nil, chain.log)
		case types.EventGetBlockExecResults:
		case types.EventGetExecutedBlock:
			go mq.ProcessMsg(msg, chain.getExecutedBlock, nil, chain.log)
		case types.EventGetExecutedBlocks:
			go mq.ProcessMsg(msg, chain.getExecutedBlocks, nil, chain.log)
		case types.EventGetBlockHeight:
			go mq.ProcessMsg(msg, chain.getBlockHeight, nil, chain.log)
		case types.EventGetLastBlock:
			go mq.ProcessMsg(msg, chain.getLastBlock, nil, chain.log)
		case types.EventTxTotalQuery:
			go mq.ProcessMsg(msg, chain.getTxTotal, nil, chain.log)
		case types.EventQueryContractTotal:
			go mq.ProcessMsg(msg, chain.getContractTotal, nil, chain.log)
		case types.EventGetBlockNumber:
			go mq.ProcessMsg(msg, chain.getMaxBlockNumber, nil, chain.log)
		case types.EventGetBlockSequence:
			go mq.ProcessMsg(msg, chain.getMaxBlockSequence, nil, chain.log)
		case types.EventGetMaxExecutedBlockNumber:
			go mq.ProcessMsg(msg, chain.getMaxExecutedBlockNumber, nil, chain.log)
		case types.EventQueryCheckedBlockTopic:
			go mq.ProcessMsg(msg, chain.getCheckedBlockTopic, nil, chain.log)
		case types.EventQueryReceivedBlockTopic:
			go mq.ProcessMsg(msg, chain.getReceivedBlockTopic, nil, chain.log)
		case types.EventRequestElectionContext:
			go mq.ProcessMsg(msg, chain.getElectionContext, nil, chain.log)
		case types.EventRequestRoundMinersInfo:
			go mq.ProcessMsg(msg, chain.getRoundMinersInfo, nil, chain.log)
		case types.EventQueryCurrentBlockSequence:
			go mq.ProcessMsg(msg, chain.getCurrentBlockSequence, nil, chain.log)
		case types.EventDelPendingPoolBlock:
			go mq.ProcessMsg(msg, chain.deletePendingBlock, nil, chain.log)
		//boss subscriber
		case types.EventSubscriber:
			go mq.ProcessMsg(msg, chain.AddSubscriber, nil, chain.log)
		case types.EventUnsubscriber:
			go mq.ProcessMsg(msg, chain.DeleteSubscriber, nil, chain.log)
		}
	}
}

// ProcTopicMsg 处理Topic模式消息
func (chain *BlockChain) ProcTopicMsg() {
	for msg := range chain.topicClient.Recv() {
		chain.log.Debugf("blockchain topic client recv msg: %s[%d]", types.GetEventName(int(msg.Ty)), msg.Ty)
		msgtype := msg.Ty
		switch msgtype {
		case types.EventAddBlock:
			go mq.ProcessMsg(msg, chain.addBlock, nil, chain.log)
		case types.EventBroadcastAddBlock:
			go mq.ProcessMsg(msg, chain.addBroadcastBlock, nil, chain.log)
		case types.EventSaveBlock:
			go mq.ProcessMsg(msg, chain.saveBlock, nil, chain.log)
		case types.EventReplyPeerInfo:
			go mq.ProcessMsg(msg, chain.updatePeerInfo, nil, chain.log)
		case types.EventStoreCheckedBlockTopic:
			go mq.ProcessMsg(msg, chain.storeCheckedBlockTopic, nil, chain.log)
		case types.EventStoreElectionContext:
			go mq.ProcessMsg(msg, chain.storeElectionContext, nil, chain.log)
		case types.EventStoreRoundMinersInfo:
			go mq.ProcessMsg(msg, chain.storeRoundMinersInfo, nil, chain.log)
		case types.EventStoreCurrentBlockSequence:
			go mq.ProcessMsg(msg, chain.storeCurrentBlockSequence, nil, chain.log)
		//worker heart beat
		case types.EventBlockWorkerHeartbeat:
			go mq.ProcessMsg(msg, chain.ProcWorkerHeartbeat, nil, chain.log)
		}
	}
}

// Close 关闭blockchain
func (chain *BlockChain) Close() {
	chain.isClosed.Store(true)
	if chain.topicClient != nil {
		chain.topicClient.Close()
	}

	if chain.workerClient != nil {
		chain.workerClient.Close()
	}
}
